package org.ricks.common.db;

import org.ricks.db.TableValueBuilder;
import org.ricks.log.Logger;

import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author chenwei
 * @Description:异步缺陷，异步一条一条的处理 。无法批量处理
 * 优化点 todo
 * @date 2020/12/2914:52
 */
public class AsyncRepositoryDao<T,K> extends RepositoryDao<T,K> {

    public AsyncRepositoryDao(Class clazz,long delayTime) {
        super(clazz);
        this.delayTime = delayTime;
        startSchedule();
    }

    public AsyncRepositoryDao(Class clazz,long delayTime,int split) {
        super(clazz);
        this.delayTime = delayTime;
        startSchedule();
    }

    private final ConcurrentLinkedQueue<Change<T, K>> waitChange = new ConcurrentLinkedQueue();
    private final ConcurrentMap<K, Save<T, K>> waitSave = new ConcurrentHashMap<>();
    private final Set<K> waitAdd = new HashSet<>();
    long delayTime = 10000;


    public AsyncRepositoryDao() {
        startSchedule();
    }

    private void startSchedule() {
        AsyncRepositoryMgr.get().getScheduledExecutor().scheduleWithFixedDelay(
                () -> delay(false),
                delayTime,
                delayTime,
                TimeUnit.MILLISECONDS);
        AsyncRepositoryMgr.get().addHook(() -> delay(true));
    }

    private synchronized void delay(boolean force) {
        try {
            while (!waitChange.isEmpty()) {
                Change<T, K> change = waitChange.poll();
                if (change.add) {
                    delayAdd(change.obj, change.pk);
                } else {
                    delayRemove(change.pk);
                }
            }
            delaySave(force);
        } catch (Throwable e) {
            Logger.error("AsyncCacheRepository delay failure. {} " ,e.getMessage(), e);
        }
    }

    private void delayAdd(T obj, K primaryKey) {
        if (waitAdd.remove(primaryKey)) {
            super.save(obj);
        }
    }

    private void delayRemove(K pk) {
        super.delete( pk);
    }

    private void delaySave(boolean force) {
        long time = System.currentTimeMillis();

        List<Save<T, K>> updates = waitSave.values().stream()
                .filter(value -> force || time - value.saveTime >= delayTime)
                .collect(Collectors.toList());

        if (!updates.isEmpty()) super.update((T[])updates.stream().map(value -> value.obj).toArray());

        updates.stream().filter(value -> force || time - value.saveTime >= delayTime).forEach(value -> waitSave.remove(value.pk));
        Logger.debug("async save table[{}]  nowUpdateCount[{}] waitSaveCount[{}]", tableMeta.getName(), updates.size(), waitSave.size());
    }

    public boolean add(T obj) {
        K primaryKey =  (K) TableValueBuilder.getPrimaryKeyValue(tableMeta, obj);
        waitAdd.add(primaryKey);
        waitChange.add(new Change<>(obj, primaryKey));
        return true;
    }

    public boolean remove(T obj) {
        return removeByPrimaryKey((K) TableValueBuilder.getPrimaryKeyValue(tableMeta, obj));
    }

    public boolean removeByPrimaryKey(K primaryKey) {
        if (waitAdd.remove(primaryKey)) {
            return true;
        }
        waitSave.remove(primaryKey); //删除，移出db更新操作
        waitChange.add(new Change<>(primaryKey));
        return true;
    }

    public boolean save(T obj) {
        Objects.requireNonNull(obj, "async cache repository save param not null");
        K pk = (K) TableValueBuilder.getPrimaryKeyValue(tableMeta, obj);
        waitSave.computeIfAbsent(pk, k -> new Save<>(obj, pk, System.currentTimeMillis()));
        return true;
    }

    @Override
    public void save(T... objects) {
        for (T t : objects) {
            add(t);
        }
    }

    @Override
    public void update(T... objects) {
        for (T t : objects) {
            save(t);
        }
    }

    @Override
    public void delete(T... objects) {
        for (T t : objects) {
            remove(t);
        }
    }


    private static class Change<T, K> {
        T obj;
        K pk;
        boolean add;
        Change(T obj, K pk) {
            this.obj = obj;
            this.pk = pk;
            this.add = true;
        }
        Change(K pk) {
            this.pk = pk;
            this.add = false;
        }
    }

    private static class Save<T, K> {
        T obj;
        K pk;
        long saveTime;
        Save(T obj, K pk, long saveTime) {
            this.obj = obj;
            this.pk = pk;
            this.saveTime = saveTime;
        }
    }

}
